﻿using Basic.EventBus.Attributes;
using Basic.EventBus.Eventbus;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System.Reflection;

namespace Basic.EventBus.EventBusRabbitMQ;

internal class RabbitMQSubscriber : IEventSubscriber
{
    private static readonly string ExchangeType = "topic";

    private List<IBasicConsumer> SubscriberInfos;
    private RabbitMQConnections Connections;

    public RabbitMQSubscriber(RabbitMQConnections connections)
    {
        Connections = connections;
        SubscriberInfos = new List<IBasicConsumer>(16);
    }

    public void AddSubscriber(ISubscriberHandler subscriberInfo)
    {
        var conn = Connections.GetConnection();
        try
        {
            if (!conn.IsOpen) throw new ConnectFailureException("IConnection.IsOpen is false;", null);

            var channel = conn.CreateModel();
            if (!channel.IsOpen) throw new ConnectFailureException("IModel.IsOpen is false;", null);

            var subscriberAttributeObj = subscriberInfo.GetType().GetCustomAttribute<EventSubscriberAttribute>();
            if (subscriberAttributeObj == null) return;

            string exchangeName = subscriberAttributeObj.TopicObj.Group;
            string queueName = subscriberAttributeObj.TopicObj.QueueName ?? subscriberAttributeObj.TopicObj.Topic;
            string routingKey = subscriberAttributeObj.TopicObj.Key;
            bool durable = subscriberAttributeObj.TopicObj.Durable;

            channel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType, true, autoDelete: false, null);
            channel.QueueDeclare(queue: queueName, durable, exclusive: false, autoDelete: !durable, null);
            channel.QueueBind(queueName, exchangeName, routingKey, null);

            ushort qos = subscriberAttributeObj.ConcurrencyNumber;
            if (qos > 0)
                channel.BasicQos(0, qos, false);

            IBasicConsumer consumer = CreateConsumer(channel, subscriberInfo, subscriberAttributeObj);
            SubscriberInfos.Add(consumer);
        }
        catch
        {
            throw;
        }
        finally
        {
            Connections.Return(conn);
        }
    }

    public void AddSubscriber(ISubscriberHandler[] subscriberInfos)
    {
        if (subscriberInfos?.Length > 0)
            foreach (var item in subscriberInfos)
                AddSubscriber(item);
    }

    private IBasicConsumer CreateConsumer(IModel channel, ISubscriberHandler subscriberInfo, EventSubscriberAttribute subscriberAttributeObj)
    {
        Action<object, BasicDeliverEventArgs> handler = async (sender, args) =>
        {
            try
            {
                byte[] data = args.Body.Span.ToArray();
                bool completed = await subscriberInfo.HandleMessage(data);
                if (!subscriberAttributeObj.AutoAck)
                {
                    if (completed)
                        channel.BasicAck(args.DeliveryTag, false);
                    else
                        channel.BasicNack(args.DeliveryTag, false, true);
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"{subscriberInfo.GetType().Name}.HandleMessage() Error:{ex.Message}");
            }
        };

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += new EventHandler<BasicDeliverEventArgs>(handler);
        channel.BasicConsume(subscriberAttributeObj.TopicObj.Topic, subscriberAttributeObj.AutoAck, consumer);
        return consumer;
    }

    #region Dispose
    private bool disposedValue;

    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                // TODO: 释放托管状态(托管对象)
            }
            // TODO: 将大型字段设置为 null
            disposedValue = true;
        }
    }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }

    #endregion

}